package com.test.study.ToolsUtils.strom.trident;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * 连接多个数据源
 * @author huyong
 *
 */
public class TridentTopologyMeger {

	@SuppressWarnings("unchecked")
	public static void main(String[] args) {
		TridentTopology topology=new TridentTopology();
		FixedBatchSpout fixedBatchSpout = new FixedBatchSpout(new Fields("sentence"), 4, new Values(1), new Values(2),
				new Values(3), new Values(4));
		fixedBatchSpout.setCycle(true);
		Stream newStream1 = topology.newStream("tuple1", fixedBatchSpout);
		Stream newStream2 = topology.newStream("tuple2", fixedBatchSpout);
		topology.merge(newStream1,newStream2).each(new Fields("sentence"), new MyTridentFilter()).each(new Fields("sentence"), new MyTridentFunction(),new Fields(""));
		//这里接受2个数据源
		LocalCluster cluster=new LocalCluster();
		cluster.submitTopology("TridentTopologyMeger", new Config(), topology.build());

		

	}

}
